package com.edu.yx.app.App_04_DwsApp.Dws_02_User;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.PropertyNamingStrategy;
import com.alibaba.fastjson.serializer.SerializeConfig;
import com.edu.yx.app.APP_01_BaseApp.BaseApp;
import com.edu.yx.bean.UserActBean;
import com.edu.yx.util.TempUtil;
import com.edu.yx.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

import static com.edu.yx.common.Constant.*;

public class Dws_01_User_Action_Count extends BaseApp {

    public static void main(String[] args) {
        new Dws_01_User_Action_Count().init(
                4021,
                2,
                "Dws_01_User_Action_Count",
                TOPIC_DWD_USER_ACT_DETAIL
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {

        //todo 测试需要修改日期，maxwell的采集日期和jar包的日期，最好间隔一个月，观察七日回流用户
        //1.数据过来按照 新增、活跃、七日回流进行封装
        SingleOutputStreamOperator<UserActBean> parseToPojo = parseToPojo(stream);

        //2.windowAll开窗聚合
        //todo 为什么迟到的原因：maxwell的时间没改成当天，导致业务数据的日期有问题
        SingleOutputStreamOperator<UserActBean> windowAndAgg = windowAndAgg(parseToPojo);

        //3.写入doris
        writeDoris(windowAndAgg);

    }

    private void writeDoris(SingleOutputStreamOperator<UserActBean> resultStream) {
        resultStream
                .map(bean ->{
                    //配置json可以将驼峰转换成带下划线，然后方便导入数据到doris
                    SerializeConfig config = new SerializeConfig();
                    config.propertyNamingStrategy = PropertyNamingStrategy.SnakeCase;

                    String json = JSON.toJSONString(bean, config);
//                    System.out.println(json);
                    return json;
                })
                .addSink(FlinkSinkUtil.getDorisSink("edu.dws_user_action_count"));

    }

    private SingleOutputStreamOperator<UserActBean> windowAndAgg(SingleOutputStreamOperator<UserActBean> parseToPojo) {

       return  parseToPojo
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<UserActBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner((bean,ts)->bean.getTs())
                )
                .windowAll(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<UserActBean>() {

                            @Override
                            public UserActBean reduce(UserActBean value1,
                                                      UserActBean value2) throws Exception {

                                value1.setActiveUserCt(value1.getActiveUserCt() + value2.getActiveUserCt());
                                value1.setNewUserCt(value1.getNewUserCt() + value2.getNewUserCt());
                                value1.setSevenBackUserCt(value1.getSevenBackUserCt() + value2.getSevenBackUserCt());

                                return value1;
                            }
                        }, new ProcessAllWindowFunction<UserActBean, UserActBean, TimeWindow>() {
                            @Override
                            public void process(Context context,
                                                Iterable<UserActBean> elements,
                                                Collector<UserActBean> out) throws Exception {

                                UserActBean bean = elements.iterator().next();
                                bean.setStt(TempUtil.toDateTime(context.window().getStart()));
                                bean.setEdt(TempUtil.toDateTime(context.window().getEnd()));
                                bean.setCurDate(TempUtil.toDate(System.currentTimeMillis()));

                                out.collect(bean);
                            }
                        }

                );
    }

    private SingleOutputStreamOperator<UserActBean> parseToPojo(DataStreamSource<String> stream) {

        return stream
                .map(strjson -> JSONObject.parseObject(strjson))
                .keyBy(json -> json.getString("user_id"))
                .process(new KeyedProcessFunction<String, JSONObject, UserActBean>() {
                    private ValueState<Long> lastDate;
                    private ValueState<String> NowDate;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        //上次登陆时间
                        lastDate = getRuntimeContext().getState(new ValueStateDescriptor<Long>("lastDate", Long.class));

                        //活跃用户登陆
                        NowDate = getRuntimeContext().getState(new ValueStateDescriptor<String>("NowDate", String.class));
                    }

                    @Override
                    public void processElement(JSONObject value,
                                               Context ctx,
                                               Collector<UserActBean> out) throws Exception {

                        Long ts = value.getLong("ts") * 1000L;
                        String today = TempUtil.toDate(ts);
                        String type = value.getString("type");

                        //todo 判断活跃用户
                        //一个用户同一天多次update，只能算一次,首次进来状态为空，第二天进来与第一天状态不一样
                        //当天用注册行为的用户也要算一次活跃
                        if (!today.equals(NowDate.value())) {

                            //如果是注册，直接对外抛
                            if ("insert".equals(type)){
                                out.collect(
                                        new UserActBean(
                                                "",
                                                "",
                                                "",
                                                1L,
                                                1L,
                                                0L,
                                                ts
                                        )
                                );
                            }


                            if ("update".equals(type)){
                                if (lastDate.value() == null || ts - lastDate.value() < 7 * 24 * 60 * 60 * 1000 ) {
                                    out.collect(
                                            new UserActBean(
                                                    "",
                                                    "",
                                                    "",
                                                    0L,
                                                    1L,
                                                    0L,
                                                    ts
                                            )
                                    );

                                }else if (ts - lastDate.value() >= 7 * 24 * 60 * 60 * 1000) {
                                    out.collect(
                                            new UserActBean(
                                                    "",
                                                    "",
                                                    "",
                                                    0L,
                                                    1L,
                                                    1L,
                                                    ts
                                            )
                                    );
                                }
                            }

                            //判断首次活跃用户，以及不满足七天回流用户

                            NowDate.update(today);
                            lastDate.update(ts);
                        }
                    }
                });
    }
}